package io.jjz;

import io.jjz.consts.MerchType;
import io.jjz.consts.OrderField;
import io.jjz.pb.MerchQuery;
import io.jjz.utils.ByteUtil;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FamilyFilter;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;

/**
 * Author       : 蒋俊钊 80256973
 * Date         : 2019/3/4
 * Copyright    (C) ChinaMerchantsBank
 */
public class MerchEndpoint extends MerchQuery.MerchService implements Coprocessor, CoprocessorService {

    private RegionCoprocessorEnvironment env;
    private byte[] dataFamily = "d".getBytes();
    private byte[] indexFamily = "i".getBytes();

    @Override
    public void getAllData(RpcController controller, MerchQuery.MerchRequest request, RpcCallback<MerchQuery.MerchResponse> done) {
        String orderName = OrderField.getNameById(request.getOrderBy()); // 排序的字段名 "YTA"
        String billDate = request.getBillDate(); // 日期
        String branchNo = request.getBranchNo(); // 分行号 "000"
        String merchType = MerchType.getIdByName(request.getMerchType().name()); //商户类型代号 "03"

        int orderTpye = Integer.parseInt(request.getOrderType()); //  0:正序   1：倒序
        int pageNum = Integer.parseInt(request.getPageNum());  // 页码从1开始
        int pageSize = Integer.parseInt(request.getPageSize());

        // 按索引区rowkey规则构建出扫描范围的起止rowkey
        byte[] startKey = ByteUtil.mergerBytes(orderName.getBytes(), billDate.getBytes(), branchNo.getBytes(), merchType.getBytes());
        byte[] endKey = ByteUtil.mergerBytes(orderName.getBytes(), billDate.getBytes(), branchNo.getBytes(), (merchType + "1").getBytes());

        Scan scan = new Scan(startKey, endKey);
        // 只扫描i列族
        FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(indexFamily));
        scan.setFilter(familyFilter);

        MerchQuery.MerchResponse response = null;
        InternalScanner scanner = null;
        try {
            scanner = env.getRegion().getScanner(scan);
            ArrayList<Cell> indexResults = new ArrayList<>();
            ArrayList<Get> getArrayList = new ArrayList<>();

            // todo: scan怎么倒序
            // 需要忽略的前n条记录
            int res = (pageNum - 1) * pageSize;
            while (scanner.next(indexResults) && pageSize > 0) {
                if (res >= indexResults.size()) {
                    res -= indexResults.size();
                    indexResults.clear();
                    continue;
                }
                for (Cell c : indexResults) {
                    byte[] indexRowkey = CellUtil.cloneRow(c);
                    // todo: bug
                    String merchNo = ByteUtil.getMerchNoFromBytes(indexRowkey);
                    // 索引对应数据的rowkey
                    String rowkey = billDate + branchNo + merchType + merchNo;
                    Get get = new Get(rowkey.getBytes());
                    get.addFamily(dataFamily);
                    getArrayList.add(get);
                    pageSize--;
                }
            }

            // 根据索引查主数据
            TableName tableName = env.getRegionInfo().getTable();
            Table table = env.getTable(tableName);
            Result[] dataResult = table.get(getArrayList);

            MerchQuery.MerchResponse.Builder builder = MerchQuery.MerchResponse.newBuilder()
                    .setDataSize(dataResult.length).setTotalCount(10);

            for (int i = 0; i < dataResult.length; i++) {
                Result r = dataResult[i];
                MerchQuery.MerchData merchData = MerchQuery.MerchData.newBuilder()
                        .setBillDate(new String(r.getValue(dataFamily, "BD".getBytes())))
                        .setMchNo(new String(r.getValue(dataFamily, "MNO".getBytes())))
                        .setMchName(new String(r.getValue(dataFamily, "MN".getBytes())))
                        .setMchType(new String(r.getValue(dataFamily, "MT".getBytes())))
                        .setBussType(new String(r.getValue(dataFamily, "BT".getBytes())))
                        .setBranchNo(new String(r.getValue(dataFamily, "BNO".getBytes())))
                        .setBranchName(new String(r.getValue(dataFamily, "BN".getBytes())))
                        .setYearTradeAmount(Bytes.toDouble(r.getValue(dataFamily, "YTA".getBytes())))
                        .setYearTradeNum(Bytes.toDouble(r.getValue(dataFamily, "YTN".getBytes())))
                        .setDailyTradeAmount(Bytes.toDouble(r.getValue(dataFamily, "DTA".getBytes())))
                        .setDailyAmountPercent(Bytes.toDouble(r.getValue(dataFamily, "DAP".getBytes())) + "%")
                        .setDailyTradeNum(Bytes.toDouble(r.getValue(dataFamily, "DTN".getBytes())))
                        .setDailyNumPercent(Bytes.toDouble(r.getValue(dataFamily, "DNP".getBytes())) + "%")
                        .setMonthTradeAmount(Bytes.toDouble(r.getValue(dataFamily, "MTA".getBytes())))
                        .setMonthAmountPercent(Bytes.toDouble(r.getValue(dataFamily, "MAP".getBytes())) + "%")
                        .setMonthNumPercent(new String(r.getValue(dataFamily, "MTN".getBytes())))
                        .setMonthNumPercent(new String(r.getValue(dataFamily, "MNP".getBytes())))
                        .setMonth2TradeAmount(Bytes.toDouble(r.getValue(dataFamily, "M2TA".getBytes())))
                        .setMonth2AmountPercent(Bytes.toDouble(r.getValue(dataFamily, "M2AP".getBytes())) + "%")
                        .setMonth2TradeNum(Bytes.toDouble(r.getValue(dataFamily, "M2TN".getBytes())))
                        .setMonth2NumPercent(Bytes.toDouble(r.getValue(dataFamily, "M2NP".getBytes())) + "%")
                        .setPBranchNo(new String(r.getValue(dataFamily, "PBNO".getBytes())))
                        .setFourthCode(new String(r.getValue(dataFamily, "FC".getBytes())))
                        .setIsRepeat(new String(r.getValue(dataFamily, "IP".getBytes())))
                        .build();
                builder.setMerchData(i, merchData);
            }
            response = builder.build();


        } catch (IOException e) {
            e.printStackTrace();
        }  finally {
            if (scanner != null) {
                try {
                    scanner.close();
                } catch (IOException e) {
                }
            }
        }
        done.run(response);
    }

    @Override
    public void test(RpcController controller, MerchQuery.MerchRequest request, RpcCallback<MerchQuery.MerchResponseTest> done) {
        MerchQuery.MerchResponseTest response = MerchQuery.MerchResponseTest.newBuilder()
                .setDataSize(99)
                .build();
        done.run(response);
    }

    @Override
    public void start(CoprocessorEnvironment env) throws IOException {
        if (env instanceof RegionCoprocessorEnvironment) {
            this.env = (RegionCoprocessorEnvironment) env;
        } else {
            throw new CoprocessorException("Must be loader on a table region!");
        }
    }

    @Override
    public void stop(CoprocessorEnvironment env) throws IOException {
    }

    @Override
    public Service getService() {
        return this;
    }
}
